package com.wuwangfu.async;


import com.wuwangfu.func.AsyncHttpFunc;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * @Description：异步 IO
 * @Author：jcshen
 * @Date：2023-06-29
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/datastream/operators/asyncio/
 *
 */
public class AsyncQueryFromHttp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        //异步IO发生REST请求地址
        String url = "http://localhost:8080/api";
        //最大异步请求数量
        int capacity = 20;
        //使用AsyncDataStream调用unordereWait方法，并传入异步请求的Function
        SingleOutputStreamOperator<Tuple2<String, String>> result = AsyncDataStream.unorderedWait(
                lines,//输入的数据流
                new AsyncHttpFunc(url, capacity),//异步查询的Function实例
                5000,//超时时间
                TimeUnit.MILLISECONDS,//时间单位
                capacity//异步请求队列最大的数量，不传该参数默认值为100
        );

        result.print();

        env.execute();

    }
}
